package net.bwie.realtime.dws.log.function;

import net.bwie.realtime.dws.log.bean.AreaControl;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import utils.DateTimeUtil;

public class TrafficViewReportWindowFunction
        implements WindowFunction<AreaControl,String,String, TimeWindow> {
    @Override
    public void apply(String s,
                      TimeWindow window,
                      Iterable<AreaControl> input,
                      Collector<String> out) throws Exception {
        //1.窗口时间
        String windowStartTime = DateTimeUtil.convertLongToString(window.getStart(), "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(window.getEnd(), "yyyy-MM-dd HH:mm:ss");

        //2.获取窗口计算结果
        AreaControl bean = input.iterator().next();
        Long carCount = bean.getCarCount();

        //3.输出
        String output = windowStartTime + "," + windowEndTime + "," + windowStartTime.substring(0, 10)
                + "," + s + "," + carCount;
        out.collect(output);
    }
}
